package com.wfh.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.wfh.config.RabbitMQConfig;
import com.wfh.listener.thumb.msg.ThumbEvent;
import com.wfh.mapper.BlogMapper;
import com.wfh.model.entity.Thumb;
import com.wfh.service.ThumbService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;

/**
 * @Author FengHuan Wang
 * @Date 2025/4/20 9:08
 * @Version 1.0
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class ThumbConsumer {

    private final BlogMapper blogMapper;

    private final ThumbService thumbService;

    /**
     * 批量处理配置
     * @param thumbEvent
     */
    @RabbitListener(queues = RabbitMQConfig.THUMB_QUEUE, containerFactory = "rabbitListenerContainerFactory")
    @Transactional(rollbackFor = Exception.class)
    public void processThumbEvent(@Payload ThumbEvent thumbEvent){
        log.info("开始处理点赞事件：{}", thumbEvent);
        
        if (thumbEvent == null) {
            log.warn("收到空消息，跳过处理");
            return;
        }
        
        // 使用一个线程安全的Map
        Map<Long, Long> countMap = new ConcurrentHashMap<>();
        // 空列表，用途存储最终生成的Thumb对象
        List<Thumb> thumbs = new ArrayList<>();

        // 并行处理消息
        LambdaQueryWrapper<Thumb> wapper = new LambdaQueryWrapper<>();
        // 这是一个线程安全的原子变量，默认值未false， 用于标记是否需要点赞
        AtomicReference<Boolean> needRemove = new AtomicReference<>(false);

        // 处理单个事件
        ThumbEvent event = thumbEvent;
        
        // 处理单个事件
        ThumbEvent.EventType finalAction = event.getType();
        if (finalAction == ThumbEvent.EventType.INCR){
            countMap.merge(event.getBlogId(), 1L, Long::sum);
            Thumb thumb = new Thumb();
            thumb.setBlogId(event.getBlogId());
            thumb.setUserId(event.getUserId());
            thumbs.add(thumb);
        }else {
            needRemove.set(true);
            wapper.or().eq(Thumb::getUserId, event.getUserId()).eq(Thumb::getBlogId, event.getBlogId());
            countMap.merge(event.getBlogId(), -1L, Long::sum);
        }

        // 批量更新数据库
        if (needRemove.get()){
            thumbService.remove(wapper);
        }
        batchUpdateBlogs(countMap);
        batchInsertThumbs(thumbs);
    }

    private void batchInsertThumbs(List<Thumb> thumbs) {
        if (!thumbs.isEmpty()){
            // 分批次插入
            thumbService.saveBatch(thumbs, 500);
        }
    }

    private void batchUpdateBlogs(Map<Long, Long> countMap) {
        if (!countMap.isEmpty()){
            blogMapper.batchUpdateThumbCount(countMap);
        }
    }

    /**
     * 死信队列处理
     * @param thumbEvent
     */
    @RabbitListener(queues = RabbitMQConfig.THUMB_DLQ_QUEUE)
    public void consumerDlq(@Payload ThumbEvent thumbEvent){
        // todo 过期消息处理
        log.info("死信队列处理：{}", thumbEvent);
        log.info("消息已入库");
        log.info("已经通知相关人员{}处理", "坤哥");
    }
}
